#ifdef CONFIG_MSPROTOCOL_MSAV
#define MSAV_C
#include <sys/types.h>
#include <sys/stat.h>		//mkdir()
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>

#ifdef OS_WIN32
#include <Windows.h>
#include <process.h>    /* _beginthread, _endthread */
#endif

#include <libavformat/avformat.h>

#if defined(OS_ANDROID)||defined(OS_LINUX_SOC)
#define HAVE_PTHREAD_FIFO 1
#include<sys/prctl.h>
#elif defined(OS_WIN32)
#define HAVE_PTHREAD_FIFO 0
#endif
#if HAVE_PTHREAD_FIFO
#include <pthread.h>
#endif
//user header
#include <libmscommon/mscommon.h>
#include <libmscommon/mssi.h>
#include <libmscommon/msfifo.h>
#include <libmscommon/msthread.h>
#include <libmscommon/mstime.h>
#include <libmslog/mslog.h>
#include "msprotocol.h"
#define FLAG "MSAV"

#define MSAV_C

#define RX_PKT_SIZE 	1316*4
typedef struct MSAVPTCContext {
	AVIOContext *ioput;
#if HAVE_PTHREAD_FIFO
	pthread_t circular_buffer_thread;
	int thread_started;
    	/* Circular Buffer variables for use in UDP receive code */
	MSFIFOBLOCKContext *pmsfifo_ctt;
	uint8_t tmp[RX_PKT_SIZE];
	int circular_buffer_error;	
#endif
} MSAVPTCContext;

#if HAVE_PTHREAD_FIFO
static void *msav_circular_buffer_taskfunc( void *_URLContext)
{
	URLContext *s = _URLContext;
	MSAVPTCContext * msav_ctt=s->priv_data;
	int left_size=0;
	int old_cancelstate;
      	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
	msthread_api_setcurname("msav_thread");
	while(1) {	
		if(0==msav_ctt->thread_started){
			return NULL;
		}
		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_cancelstate);
		int len= avio_read(msav_ctt->ioput, msav_ctt->tmp,RX_PKT_SIZE);
		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
		if (len <0) {
			if (ms_neterrno != ms_averror(EAGAIN) 
				&& ms_neterrno != ms_averror(EINTR)) {
				msav_ctt->circular_buffer_error = ms_neterrno;
				ms_errret(ms_null,"circular_buffer_error:(%d,%s)",msav_ctt->circular_buffer_error,ms_strerr);
			}
			ms_usleep(500);
			continue;
        		}else if(len ==0){
			ms_usleep(500);
			continue;
		}
recheck:
		if(0==msav_ctt->thread_started){
			return NULL;
		}
		left_size=msfifoblock_api_sizeleft(msav_ctt->pmsfifo_ctt);
		if( left_size<len) {/* No Space left */
			
			ms_usleep(500);
			goto recheck;
		}
		msfifoblock_api_in(msav_ctt->pmsfifo_ctt, msav_ctt->tmp, len, NULL);
		ms_usleep(500);
	}
	return NULL;
}
#endif
int msav_api_openinterruptcallback(void *arglist) {
	int * timeout=arglist;
	return (* timeout);
}
void * msav_api_openinouttime(void * arglist)
{
	int * timeout=arglist;
	time_t timep=time(NULL);
	while(1){
		if(mstime_api_counter_async(&timep,4)){
			(* timeout)=1;
			return ms_null;
		}
		ms_msleep(50);
		if(1==(* timeout)){
			return ms_null;
		}
	}
}
static int msav_open(URLContext *s)
{
	if(1==s->is_connected){
		return 0;
	}
	int ret=0;
	AVIOContext *ioput = NULL;
	byte name_mac[mscfg_maxlen_url+128]={0};
	ms_sprintfs(name_mac, "MSAVPTCContext(%s)", s->url);
	MSAVPTCContext * msav_ctt=s->priv_data=(void *)ms_malloc(sizeof(MSAVPTCContext),name_mac);
	if(NULL==s->priv_data){
		ms_errgoto(fail,"ms_malloc MSAVPTCContext faild(%d,%s)",errno,strerror(errno));
	}
	memset(msav_ctt,0,sizeof(MSAVPTCContext));
	
	int flags=0;
	if(!(s->flags&FLAG_WRITE)){	
		flags=AVIO_FLAG_READ|AVIO_FLAG_NONBLOCK;
	}else if(s->flags&FLAG_WRITE){
		flags=AVIO_FLAG_WRITE;
	}
	
	int timeout=0;
	#if defined(OS_ANDROID)||defined(OS_LINUX_SOC)
		pthread_t handle_openin;
		pthread_create(&handle_openin, NULL, (void * (*)(void *))msav_api_openinouttime,&timeout);		
	#elif defined(OS_WIN32)
		unsigned long handle_openin=_beginthreadex(NULL, 0, msav_api_openinouttime, &timeout, 0, NULL);
	#endif
	ms_debug( "open input file:%s, flags:%#x,s->flags:%#x",s->url,flags,s->flags);
	/* BEGIN: Added by su<sugao_cn@163.com>, 2016/2/19   PN:RTSP options */
	AVDictionary *options=NULL;
	//av_dict_set(&options, "rtsp_transport","tcp", 0);
	//av_dict_set(&options, "probesize","10000000", 0);
	
	/* END:   Added by su<sugao_cn@163.com>, 2016/2/19 */
	AVIOInterruptCB int_cb;
	int_cb.callback=msav_api_openinterruptcallback;
	int_cb.opaque=&timeout;
	ret = avio_open2(&ioput, s->url, flags, &int_cb, &options);
	timeout=1;
	ms_msleep(200);
	#if defined(OS_ANDROID)||defined(OS_LINUX_SOC)
		#if defined(OS_LINUX_SOC)
			pthread_cancel(handle_openin);
		#endif
		if (pthread_join(handle_openin, NULL) != 0){
			ms_error("pthread_join(): error");
		}	
	#elif defined(OS_WIN32)	
		CloseHandle((HANDLE)handle_openin);
	#endif	
	if (ret < 0) {
		ms_errgoto(fail,"Failed to open input: %s: %s.", s->url,av_err2str(ret));
	}	


	msav_ctt->ioput=ioput;
	
#if HAVE_PTHREAD_FIFO
	if(!(s->flags&FLAG_WRITE)&&(s->flags&FLAG_USE_THREAD_FIFO)){
		if(0==s->urlctt_opt.msav_opt.fifo_size){
			s->urlctt_opt.msav_opt.fifo_size = 7*4096;
			ms_waring("Not set fifo_size,use default val:%d",(s->urlctt_opt.msav_opt.fifo_size));
		}
		ms_debug( "udp use thread and pmsfifo_ctt-----circular_buffer_taskfunc(fifo_size:%d)",s->urlctt_opt.msav_opt.fifo_size);
		msav_ctt->pmsfifo_ctt = msfifoblock_api_init(s->urlctt_opt.msav_opt.fifo_size,"msav");
		if (msav_ctt->pmsfifo_ctt == NULL) {
			ms_errgoto(mutex_fail, "msfifoblock_api_init failed : %s\n", strerror(ret));
		}	
		ret = pthread_create(&msav_ctt->circular_buffer_thread, NULL, msav_circular_buffer_taskfunc, s);
		if (ret != 0) {
			ms_errgoto(thread_fail, "pthread_create failed : %s\n", strerror(ret));
		}
		msav_ctt->thread_started = 1;
	}
#endif

	s->is_connected=1;	
	return 0;
	
#if HAVE_PTHREAD_FIFO
	if(!(s->flags&FLAG_WRITE)&&(s->flags&FLAG_USE_THREAD_FIFO)){
 thread_fail:
 cond_fail:
 mutex_fail:
		msfifoblock_api_deinit(&msav_ctt->pmsfifo_ctt);	
	}
#endif

fail:
	if(NULL!=msav_ctt){	
		if(NULL!=msav_ctt->ioput){
			avio_close(msav_ctt->ioput);
		}
		ms_free(msav_ctt);
		msav_ctt=NULL;
	}
	s->is_connected=0;
	return -1;
}
static int msav_read(URLContext *s ,  unsigned char *buf, int size)
{
	if(0==s->is_connected){
		return -1;
	}	
	MSAVPTCContext * msav_ctt=s->priv_data;
#if HAVE_PTHREAD_FIFO
	if(!(s->flags&FLAG_WRITE)&&(s->flags&FLAG_USE_THREAD_FIFO)){
		ms_u32 fifo_data_size=0;
		if (msav_ctt->pmsfifo_ctt) {
			do {
				
				fifo_data_size =msfifoblock_api_sizeused(msav_ctt->pmsfifo_ctt);
				if (fifo_data_size && ms_f32!=fifo_data_size) { 
					if(!(fifo_data_size<size)){
						fifo_data_size=size;
					}
					msfifoblock_api_out(msav_ctt->pmsfifo_ctt, buf, fifo_data_size, NULL);
					return fifo_data_size;
				}else if(msav_ctt->circular_buffer_error){
					int circular_buffer_error=msav_ctt->circular_buffer_error;
					return circular_buffer_error;
				}else {
					return AVERROR(EAGAIN);
            			}
			} while( 1);
    		}
	}
#endif
	return avio_read(msav_ctt->ioput, buf,size);
}
static int msav_write(URLContext *s,  unsigned char *buf, int size)
{
	if(0==s->is_connected){
		return -1;
	}
	MSAVPTCContext * msav_ctt=s->priv_data;
	avio_write(msav_ctt->ioput, buf,size);
	return 0;
}

static int msav_close(URLContext *s)
{
	if(0==s->is_connected){
		return 0;
	}
	MSAVPTCContext * msav_ctt=s->priv_data;	
#if HAVE_PTHREAD_FIFO
	if((!(s->flags&FLAG_WRITE))&&(s->flags&FLAG_USE_THREAD_FIFO)){
		if (msav_ctt->thread_started) {
			msav_ctt->thread_started=0;
		#if defined(OS_LINUX_SOC)	
			pthread_cancel(msav_ctt->circular_buffer_thread);
		#endif
			int ret = pthread_join(msav_ctt->circular_buffer_thread, NULL);
			if (ret != 0){
				ms_error( "pthread_join(): %s\n", strerror(ret));
			}
		}
		msfifoblock_api_deinit(&msav_ctt->pmsfifo_ctt);
	}	
#endif
	if(NULL!=msav_ctt){		
		avio_close(msav_ctt->ioput);
		ms_free(msav_ctt);
		msav_ctt=NULL;
	}

	s->is_connected=0;
	return 0;
}

URLProtocol ffurl_msav_protocol={
    .name		=PTLNAME_MSAV,
    .longname	=PTLDES_MSAV,
    .url_open	= msav_open,
    .url_read	= msav_read,
    .url_write	= msav_write,
    .url_close	= msav_close,
    .next	=NULL,
};
#undef MSAV_C
#endif
